import org.apache.hadoop.util.bloom.{BloomFilter, Key}
import org.apache.spark.SparkConf
import org.apache.spark.serializer.KryoSerializer
import org.apache.spark.sql.SparkSession

/**
 *
 * 在spark中使用bloom过滤器来判断uid是否已存在于历史中的示例代码
 *
 * 知识点： 布隆过滤器（原理，应用场景）
 *         序列化（jdk序列化，hadoop的writable序列化，kryo序列化）
 *
 */
object SparkUseBloomFilterDemo {


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    // 设置spark内部使用的序列化器为： kryo序列化器
    // 如果不设置，那么spark默认使用jdk的序列化器（ObjectOutputStream，而这个类对被序列化的对象要求实现Serializable标记接口)
    conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    // 如果显式注册了需要用kryo序列化的类型，那么kryo在序列化对象时，不需要为每个对象都带上类的元信息，效率更高
    conf.registerKryoClasses(Array(classOf[BloomFilter],classOf[Person]))

    val spark = SparkSession.builder().config(conf)
      .appName("布隆过滤器示例")
      .master("local")
      .getOrCreate()
    import spark.implicits._

    // 加载历史上出现过的所有deviceid
    val his = spark.createDataset(Seq("deviceid_01", "deviceid_05", "deviceid_15", "deviceid_35", "deviceid_66"))
    // 收集历史id数据到driver段
    val hisIds = his.collect()

    // 构造一个布隆过滤器，并将所有历史id映射到过滤器中
    val filter = new BloomFilter(200000000, 4, 1)
    for(id <- hisIds){
      filter.add(new Key(id.getBytes()))
    }

    // 广播
    val bc = spark.sparkContext.broadcast(filter)


    // 处理当天日志，标记新老访客
    val log = spark.createDataset(Seq("deviceid_01", "deviceid_05", "deviceid_15", "deviceid_25", "deviceid_16"))
    val res = log.map(s=>{
      val f = bc.value

      var isnew = 1
      if(f.membershipTest(new Key(s.getBytes()))) isnew = 0

      (s,isnew)
    }).toDF("device_id","isnew")


    res.show(100,false)

    spark.close()

    /**
     *
     * 这里是为了测试gitee添加的内容
     *
     *
     *
     */
  }
}
